package com.study.spark.scala.rdd

import org.apache.spark.{SparkConf, SparkContext}

/**
  * RDD:弹性分布式数据集，是一个基本的抽象，操作RDD就行操作本地集合一样，降低编程复杂度。
  * RDD算子分为Transformation 和 Action
  * RDD并不存储真正要计算的数据，而是记录RDD的转换关系（调用了什么方法，传入什么函数）
  * 特点
  * 1）一系列分区，分区有编号，有顺序
  * 2）每个分区都有一个函数作用在上面
  * 3）RDD之间存在依赖关系
  * 4）可选（key,value）类型：存在分区器，shuffle时会使用，默认是hash partitioner
  * 5）可选（从hdfs中读取数据）：可以得到数据的优先位置
  * @author: stephen.shen
  * @create: 2019-03-14 11:19
  */
object RDDDemo {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setMaster("local[*]")
      .setAppName("RDDDemo")
    val sc = new SparkContext(conf)
    // 可以从本地、hdfs等文件系统读取数据
    //val rdd = sc.textFile("")
    // 也可以从scala集合中读取（常用于测试）
    val rdd = sc.parallelize(List("a c d", "b d a", "c b d", "d a a", "d b d"))
    // word count
    rdd.flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)
      // groupByKey和reduceByKey区别，
      // groupByKey会先进行分组，再进行聚合
      // reduceByKey会先进行局部聚合，再分组，再聚合，效率会高一些
      //.groupByKey().mapValues(_.sum)
      .sortBy(_._1, true)
      .foreach(println)

    //val rdd1 = sc.parallelize(List(("a",1),("b",0),("b",2),("c",3)))
    //val rdd2 = sc.parallelize(List(("b",1),("c",3),("e",4)))

    /**
    // 并集
    println("============= 并集 ==============")
    rdd1.union(rdd2).foreach(println)

    // 交集
    println("============= 交集 ==============")
    (rdd1 intersection rdd2).foreach(println)
    // 结果：(c,3)
    **/

    /**
    // join，按照key进行join
    println("============= join ==============")
    rdd1.join(rdd2).foreach(println)
    println("============= left outer join ==============")
    rdd1.leftOuterJoin(rdd2).foreach(println)
    println("============= right outer join ==============")
    rdd1.rightOuterJoin(rdd2).foreach(println)
    **/

    // 按key分组
    //rdd1.union(rdd2).groupByKey().foreach(println)

    // 对两个RDD中的KV元素，每个RDD中相同key中的元素分别聚合成一个集合。
    //rdd1.cogroup(rdd2).foreach(println(_))

    // 笛卡尔积
    //rdd1.cartesian(rdd2).foreach(println(_))

    // Scala集合数据创建RDD，指定分区数，默认为分配的核数
    //val rdd3 = sc.parallelize(List(("a", 1), ("b", 0), ("b", 2), ("c", 3), ("b", 0), ("b", 2), ("c", 3)), 1)
    //val rdd3 = sc.parallelize(List(("a", 1), ("b", 0), ("b", 2), ("c", 3), ("b", 0), ("b", 2), ("c", 3)), 3)
    //val rdd3 = sc.parallelize(List(("a", 1), ("b", 0), ("b", 2), ("c", 3), ("b", 0), ("b", 2), ("c", 3)))
    //println(rdd3.partitions.length)
    // 从hdfs中读取数据，并且设置最小分区数为1，RDD分区数为？
    // 如果不设置最小分区数，默认传入2，RDD分区数大于等于输入的切片数
    //val rdd5 = sc.textFile("hdfs://localhost:9000/bigdata-study/spark")
    //println(rdd5.partitions.length)



    // action
    val rdd3 = sc.parallelize(List(("a", 1), ("b", 0), ("b", 2), ("c", 3), ("b", 0), ("b", 2), ("c", 3)))
    val rdd4 = sc.parallelize(List(10,1,2,3,4,5))
    val rdd5 = sc.parallelize(List(("a","1 2"),("b","3 4")))
    // collect|count|top|take等会收集到driver端
    rdd4.collect()
    rdd4.reduce(_+_)
    rdd4.count()
    rdd4.top(2)
    rdd4.take(2)
    rdd4.first()
    rdd4.takeOrdered(3)
    rdd4.saveAsTextFile("")
    rdd3.countByKey()
    // 在executor端执行打印
    rdd4.foreach(println(_))
    // 针对整个分区数据进行操作，常用于写数据库
    rdd4.foreachPartition(it=>it.foreach(println))

    rdd4.aggregate(0)(_+_,_+_)




    // Transformation
    // 底层全部调用combineByKeyWithClassTag
    rdd3.foldByKey(0)(_+_)
    rdd3.reduceByKey(_+_)
    rdd3.aggregateByKey(0)(_+_,_+_)
    rdd3.combineByKey(x=>x,(m:Int,n:Int)=>m+n,(x:Int,y:Int)=>x+y)

    rdd3.filterByRange("a","b")
    rdd5.flatMapValues(_.split(" "))


    //val rdd6 = sc.parallelize(List(1,2,3,4,5,6,7,8,9),2)
    //val rdd7 = sc.parallelize(List("a","b","c","d"),2)
    //val rdd8 = sc.parallelize(List("a","bc","def",""),2)
    // 一次拿一个分区，并且得到分区的编号
    //rdd6.mapPartitionsWithIndex((i, it:Iterator[Int]) => {
    //  it.map(x => s"分区编号：$i，值：$x")
    //}).foreach(println(_))

    // 第一个_+_分区内聚合，第二个_+_全局聚合，初始值在局部和全局都会起作用
    // 结果为：48
    //println(rdd6.aggregate(1)(_+_,_+_))
    // ||ab|cd  或者 ||cd|ab
    //println(rdd7.aggregate("|")(_+_,_+_))
    // 01 或者 10
    //println(rdd8.aggregate("")((x,y)=>math.min(x.length,y.length).toString,(x,y)=>x+y))

    val rdd9 = sc.parallelize(List(("a", 1), ("b", 1), ("b", 1), ("c", 1), ("b", 1), ("b", 1), ("c", 1)),2)
    // 按照key进行聚合，第一个_+_分区内聚合，第二个_+_全局聚合，初始值在局部起作用
    // a只在第一个分区，b在两个分区，c在第二个分区，结果如下：
    // (b,204)
    // (a,101)
    // (c,102)
    rdd9.aggregateByKey(100)(_+_,_+_).foreach(println(_))


    sc.stop()

  }
}
